package com.ds.infrastructure.hyperspace.container.internal.opentracing.reporter;

import com.google.common.collect.Maps;
import com.ds.infrastructure.hyperspace.container.internal.util.HttpRequestUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.internal.StringUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.net.URI;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import static com.ds.infrastructure.hyperspace.container.internal.opentracing.reporter.OpentracingSendChannelInitializer.HTTP_CLIENT_HANDLER_NAME;


/**
 * @Author: yaozh
 * @Description:
 */
@Slf4j
public class OpentracingSender {

    private static final int MAX_FAST_CONNECTION_COUNT = 10;
    private static final int MAX_SEND_RETRY_COUNT = 2;
    private EventLoopGroup workerGroup;
    private Bootstrap b = new Bootstrap();
    private Channel channel;
    private int connectionFailCount = 0;
    @Getter
    private String url;
    @Getter
    @Setter
    private boolean isHealth = false;
    /**
     * 配置版本，当reload的时候更新配置版本信息
     */
    private String version = UUID.randomUUID().toString();
    private CountDownLatch connectionAwaitLatch = new CountDownLatch(1);
    private boolean isShutdown = false;

    public OpentracingSender(EventLoopGroup workerGroup, String url) {
        this.workerGroup = workerGroup;
        this.url = url;
        init();
        connection();
    }

    private void init() {
        b.group(workerGroup);
        b.channel(NioSocketChannel.class);
        b.option(ChannelOption.TCP_NODELAY, true);
        b.option(ChannelOption.SO_REUSEADDR, true);
        b.option(ChannelOption.SO_KEEPALIVE, true);
        b.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(1 * 1024 * 1024, 10 * 1024 * 1024));
        b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000);
    }

    public void connection() {
        if (StringUtil.isNullOrEmpty(url)) {
            return;
        }
        URI uri = createUri(url);
        if (StringUtils.isBlank(uri.getHost())) {
            throw new IllegalStateException("Opentracing url error, hostname can't be null ! url:" + url);
        }
        b.handler(new OpentracingSendChannelInitializer(HttpRequestUtil.getSslContext(uri.getScheme())));
        b.connect(uri.getHost(), HttpRequestUtil.getPort(uri)).addListener(new ConnectFutureListener(this));
    }

    public void reload(String url) {
        this.url = url;
        this.version = UUID.randomUUID().toString();
        connection();
    }

    /**
     * send span data
     *
     * @param data
     * @return if true  auto release data
     * @throws InterruptedException
     */
    public boolean send(ByteBuf data) throws InterruptedException {
        if (!isEnable() || data == null) {
            return false;
        }
        for (int i = 0; i < MAX_SEND_RETRY_COUNT; i++) {
            if (isAvailableChannel()) {
                Map<CharSequence, Object> header = Maps.newHashMapWithExpectedSize(2);
                header.put(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
                header.put(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
                DefaultFullHttpRequest request = HttpRequestUtil.createFullHttpRequest(createUri(url), HttpVersion.HTTP_1_1, HttpMethod.POST, header, data);
                channel.writeAndFlush(request);
                return true;
            }
            if (connectionFailCount > MAX_FAST_CONNECTION_COUNT) {
                return false;
            }
            connectionAwaitLatch.await(2, TimeUnit.SECONDS);
        }
        return false;

    }

    private URI createUri(String url) {
        URI uri = HttpRequestUtil.createUri(url);
        if (!HttpScheme.HTTP.toString().equalsIgnoreCase(uri.getScheme())
                && !HttpScheme.HTTPS.toString().equalsIgnoreCase(uri.getScheme())) {
            throw new UnsupportedOperationException("Only HTTP(S) is supported.");
        }
        return uri;
    }

    public boolean isEnable() {
        return !StringUtil.isNullOrEmpty(url);
    }

    public boolean isAvailableChannel() {
        if (channel != null && channel.isActive() && channel.isWritable() && isHealth) {
            return true;
        }
        if (connectionAwaitLatch.getCount() == 0) {
            connectionAwaitLatch = new CountDownLatch(1);
        }
        return false;
    }


    public void reconnection() {
        workerGroup.schedule(() -> {
            try {
                connection();
            } catch (Exception e) {
                log.error("opentracing sender connection fail!", e);
            }

        }, getIntervalTime(), TimeUnit.SECONDS);
    }

    private Long getIntervalTime() {
        if (connectionFailCount > MAX_FAST_CONNECTION_COUNT) {
            return 3L;
        }
        if (!isHealth && connectionFailCount == 0) {
            return 3L;
        }
        return 0L;
    }

    public void connectionAwaitLatchCountDown() {
        connectionAwaitLatch.countDown();
    }

    public String getVersion() {
        return this.version;
    }

    private static class ConnectFutureListener implements GenericFutureListener<ChannelFuture> {

        private final OpentracingSender opentracingSender;

        public ConnectFutureListener(OpentracingSender opentracingSender) {
            this.opentracingSender = opentracingSender;
        }

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (opentracingSender.isShutdown) {
                return;
            }
            Channel newChannel = future.channel();
            if (!future.isSuccess()) {
                log.warn("Opentracing Connection fail! url:{},  fail count:{}", opentracingSender.url, opentracingSender.connectionFailCount);
                opentracingSender.connectionFailCount++;
                if (newChannel != null && newChannel.isOpen()) {
                    newChannel.flush().close();
                }
                opentracingSender.reconnection();
                return;
            }
            opentracingSender.channel = newChannel;
            //reset connection fail count
            opentracingSender.connectionFailCount = 0;
            opentracingSender.connectionAwaitLatch.countDown();
            opentracingSender.setHealth(false);
            OpentracingSendChannelHandler httpClientHandler = (OpentracingSendChannelHandler) newChannel.pipeline().get(HTTP_CLIENT_HANDLER_NAME);
            httpClientHandler.attachSender(opentracingSender);
            //health check
            newChannel.writeAndFlush(opentracingSender.ping());

        }
    }

    /**
     * 心跳包-维持连接
     *
     * @return
     */
    public FullHttpRequest ping() {
        if (StringUtil.isNullOrEmpty(url)) {
            return null;
        }
        Map<CharSequence, Object> header = Maps.newHashMapWithExpectedSize(2);
        header.put(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        header.put(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
        return HttpRequestUtil.createFullHttpRequest(createUri(url), HttpVersion.HTTP_1_1, HttpMethod.POST, header);
    }

    public void shutdown() {
        isShutdown = true;
    }
}
